package com.codejiwei.flink.practice;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author jiwei
 * @description
 * @date 2023/5/24 13:53
 */
public class Flink_ProcessFunction_CoProcessFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Integer> intStream = env.fromElements(1, 2, 3, 4, 5);
        DataStreamSource<String> stringStream = env.fromElements("a", "b", "c");

        intStream.connect(stringStream)
                .process(new CoProcessFunction<Integer, String, Tuple2<String, String>>() {
//                    Integer tmp;
//                    String tmpString;

                    @Override
                    public void processElement1(Integer value, Context ctx, Collector<Tuple2<String, String>> out) throws Exception {
                        System.out.println("process element1--------");
                        out.collect(Tuple2.of("process1", value.toString()));
                    }

                    @Override
                    public void processElement2(String value, Context ctx, Collector<Tuple2<String, String>> out) throws Exception {
                        System.out.println("process element2----------");
                        out.collect(Tuple2.of(value, "process2"));
                    }
                })
                .print();


//        intStream.connect(stringStream)
//                .process(new CoProcessFunction<Integer, String, String>() {
//                    @Override
//                    public void processElement1(Integer value, Context ctx, Collector<String> out) throws Exception {
//                        out.collect(value.toString());
//                    }
//
//                    @Override
//                    public void processElement2(String value, Context ctx, Collector<String> out) throws Exception {
//                        out.collect(value);
//                    }
//                })
//                .print();

        env.execute();
    }
}
